package com.ngkj.zmfptes.mq;

import com.alibaba.fastjson.JSONArray;
import com.google.gson.JsonArray;
import com.ngkj.zmfptCommon.common.bean.RpcRequestMessage;

import com.ngkj.zmfptes.bean.RentalResponseEntity;
import com.ngkj.zmfptes.service.RentalInfoService;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.tomcat.util.threads.ThreadPoolExecutor;
import org.springframework.beans.BeansException;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.*;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

/**
 * @author ChenHN
 * @since 2021/11/8 creation
 **/
@Slf4j
@Component
public class KafkaConsumer {

    @Resource
    private ApplicationContext applicationContext;

    private static ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(
            2, Runtime.getRuntime().availableProcessors(),
            1, TimeUnit.SECONDS,new LinkedBlockingQueue<>(40),
            Executors.defaultThreadFactory(),
            new java.util.concurrent.ThreadPoolExecutor.DiscardPolicy()
            );

    @KafkaListener(topics = {"zmfpt-es"},groupId = "zmfpt-id")
    public void onMessage2(ConsumerRecord<?, ?> record) {
//        threadPoolExecutor.execute(()->{
            Optional<?> kafkaMessage = Optional.ofNullable(record.value());
            log.info(">>>>>>>>>>> record = " + kafkaMessage);
            if (kafkaMessage.isPresent()) {
                RpcRequestMessage message = (RpcRequestMessage) kafkaMessage.get();
                try {
                    String interfaceName = message.getInterfaceName();
                    Class<?> aClass = Class.forName(interfaceName);
                    Object[] parameterValue = message.getParameterValue();
                    String josnString = (String) parameterValue[0];
                    Object service = applicationContext.getBean(aClass);
                    Method method = getMethod(interfaceName, message.getMethodName(), message.getParameterTypes()[0]);
                    Object invoke = method.invoke(service, josnString);
                } catch (IllegalAccessException e) {
                    e.printStackTrace();
                } catch (InvocationTargetException e) {
                    e.printStackTrace();
                } catch (ClassNotFoundException e) {
                    e.printStackTrace();
                }
            }
//        });
    }


    private Method getMethod(String className,String methodName,Class<?> propertiesClass) throws ClassNotFoundException {
        Class<?> aClass = Class.forName(className);
        Method[] methods=aClass.getMethods();
        for (Method method : methods) {
            if(method.getName().equals(methodName)){
                if(method.getParameterCount() == 1){
                    //该函数的参数类型和传入参数类型相同
                    if(method.getParameterTypes()[0].getTypeName().equals(propertiesClass.getTypeName())){
                        return method;
                        //该函数的参数类型是传入参数类型的父类
                    }else if(method.getParameterTypes()[0].getTypeName().equals(propertiesClass.getSuperclass().getTypeName())){
                        return method;
                    }else
                    {
                        Set<String> superClassAndSuperInterfaceList= this.getAllSuperClassAndSuperInterface(propertiesClass);
                        //如果传入参数类型是参数类型的子类 也返回改函数
                        if(superClassAndSuperInterfaceList.contains(method.getParameterTypes()[0].getTypeName()))
                            return method;

                    }
                }
            }
        }
        return  null;
    }

    /**
     * 获取所有父类类型和父类接口类型
     * @param clazz
     * @return
     */
    private Set<String> getAllSuperClassAndSuperInterface(Class<?> clazz){
        Set<String> superClassAndSuperInterfaceList = new HashSet<>();
        getAllSupersClasss(superClassAndSuperInterfaceList,clazz);
        getAllSuperInterfaces(superClassAndSuperInterfaceList,clazz);
        return superClassAndSuperInterfaceList;
    }

    /**
     * 递归获取所父类 类型
     * @param parentClassList
     * @param clazz
     */
    private Set<String> getAllSupersClasss(Set<String> parentClassList,Class<?> clazz){
        parentClassList.add(clazz.getSuperclass().getName());
        if(Object.class.getTypeName()!=clazz.getSuperclass().getTypeName()){
            //父类也可能实现接口
            getAllSuperInterfaces(parentClassList,clazz.getSuperclass());
            //递归查询父类
            getAllSupersClasss(parentClassList,clazz.getSuperclass());
        }
        return parentClassList;
    }

    /**
     * 递归获取父类接口
     * @param parentInterfaceList
     * @param clazz
     */
    private Set<String> getAllSuperInterfaces(Set<String> parentInterfaceList,Class<?> clazz){
        for (Class<?> aClass : clazz.getInterfaces()) {
            parentInterfaceList.add(aClass.getTypeName());
            //递归查询实现接口
            getAllSuperInterfaces(parentInterfaceList,aClass);
        }
        return parentInterfaceList;
    }

}
